package com.knight.hadoop.day12.zuoye;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 日志解析第二步 通过第一步的问题获取Pageviews表 按照session汇聚信息
 * 
 * @author
 *
 */
public class LogParseStepTwo {

	static class LogParseStepTwoMapper extends Mapper<LongWritable, Text, LogBean2, LogBean2> {

		static LogBean2 kv = new LogBean2();

		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String line = value.toString();
			String[] splitStr = line.split("\001");
			if (splitStr.length >= 9) {
				String date = splitStr[0];
				String ip = splitStr[1];
				String remoteUser = splitStr[2];
				String sessionId = splitStr[3];
				String requestURL = splitStr[4];
				String referal = splitStr[5];
				String status = splitStr[6];
				String bodyBytesSend = splitStr[7];
				String userAgent = splitStr[8];
				kv.set(date, ip, remoteUser, sessionId, requestURL, referal, status, bodyBytesSend, userAgent);
				context.write(kv, kv);
			}
		}
	}

	static class LogParseStepTwoReducer extends Reducer<LogBean2, LogBean2, Text, NullWritable> {

		static NullWritable v = NullWritable.get();
		static Text k = new Text();

		@Override
		protected void reduce(LogBean2 key, Iterable<LogBean2> values, Context context)
				throws IOException, InterruptedException {

			List<LogBean2> listBean = new ArrayList<>();

			for (LogBean2 bean : values) {
				LogBean2 logBean2 = new LogBean2();
				try {
					BeanUtils.copyProperties(logBean2, bean);
				} catch (IllegalAccessException e) {
					e.printStackTrace();
				} catch (InvocationTargetException e) {
					e.printStackTrace();
				}
				listBean.add(logBean2);
			}

			long stayTime = 10;
			int stepNum = 1;

			if (listBean != null && listBean.size() > 0) {
				if (listBean.size() == 1) {
					LogBean2 logBean2 = listBean.get(0);
					k.set(logBean2.getSessionId() + "\001" + logBean2.getIp() + "\001" + logBean2.getRemoteUser()
							+ "\001" + logBean2.getDate() + "\001" + logBean2.getRequestURL() + "\001" + stepNum
							+ "\001" + stayTime + "\001" + logBean2.getReferal() + "\001" + logBean2.getUserAgent()
							+ "\001" + logBean2.getBodyBytesSend() + "\001" + logBean2.getStatus());
					context.write(k, v);
				} else {
					for (int i = 0; i < listBean.size(); i++) {
						if (i == 0) {
							// 第一个不输出
							continue;
						}
						LogBean2 currentBean = listBean.get(i);
						LogBean2 preBean = listBean.get(i - 1);

						// 比较访问间隔时间，如果大于等于30分钟返回true
						stayTime = Utils.compareDateSec(currentBean.getDate(), preBean.getDate());

						k.set(preBean.getSessionId() + "\001" + preBean.getIp() + "\001" + preBean.getRemoteUser()
								+ "\001" + preBean.getDate() + "\001" + preBean.getRequestURL() + "\001" + stepNum
								+ "\001" + stayTime + "\001" + preBean.getReferal() + "\001" + preBean.getUserAgent()
								+ "\001" + preBean.getBodyBytesSend() + "\001" + preBean.getStatus());

						context.write(k, v);
						stepNum++;

						if (i == listBean.size() - 1) {
							// 最后一个
							stayTime = 10;
							k.set(currentBean.getSessionId() + "\001" + currentBean.getIp() + "\001"
									+ currentBean.getRemoteUser() + "\001" + currentBean.getDate() + "\001"
									+ currentBean.getRequestURL() + "\001" + stepNum + "\001" + stayTime + "\001"
									+ currentBean.getReferal() + "\001" + currentBean.getUserAgent() + "\001"
									+ currentBean.getBodyBytesSend() + "\001" + currentBean.getStatus());
							context.write(k, v);
						}
					}
				}
			}
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		job.setJarByClass(LogParseStepTwo.class);
		job.setMapperClass(LogParseStepTwoMapper.class);
		job.setReducerClass(LogParseStepTwoReducer.class);
		job.setGroupingComparatorClass(GroupLogSessionComparator.class);

		job.setMapOutputKeyClass(LogBean2.class);
		job.setMapOutputValueClass(LogBean2.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		FileInputFormat.setInputPaths(job, new Path("E:\\home\\webLog\\output1"));
		FileOutputFormat.setOutputPath(job, new Path("E:\\home\\webLog\\output2"));
		//FileInputFormat.setInputPaths(job, new Path("C:\\application\\book\\Hadoop\\output\\weblog\\step1"));
		//FileOutputFormat.setOutputPath(job, new Path("C:\\application\\book\\Hadoop\\output\\weblog\\step2"));
		
		boolean res = job.waitForCompletion(true);

		System.exit(res ? 0 : 1);
	}

}
